package com.dfire.mtm.task;

import com.dfire.mtm.constant.TaskLife;
import com.dfire.mtm.taskNode.AbstractTaskNode;
import com.dfire.mtm.taskNode.TaskNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;

/**
 * 描述:
 * 任务实现
 *
 * @author sutang
 * @create 2018-10-23 下午7:00
 */
public abstract class AbstractTaskManager implements TaskManager {

    final Logger logger = LoggerFactory.getLogger(getClass());

    protected String name;

    protected volatile boolean isEnable;

    protected final Set<TaskNode> taskNodes = new HashSet<>();

    protected final TaskNode firstNode;

    protected ExecutorService taskNodeExecutor;

    protected ExecutorService taskPollExecutor = Executors.newSingleThreadExecutor();

    protected final BlockingQueue<String> tasks = new LinkedBlockingQueue<>(10);

    protected ConcurrentHashMap<String, BlockingQueue> dataMap = new ConcurrentHashMap<>();


    public AbstractTaskManager(TaskNode firstNode) {
        this.firstNode = firstNode;
        findAllTaskNode(firstNode);
    }

    private void findAllTaskNode(TaskNode firstNode){
        taskNodes.add(firstNode);
        if(!firstNode.getChildrenNode().isEmpty()){
            for(TaskNode taskNode:firstNode.getChildrenNode()){
                findAllTaskNode(taskNode);
            }
        }
    }

    @Override
    public String getName() {
        return this.name;
    }


    @Override
    public synchronized void start() {
        if (isEnable) {
            throw new RuntimeException("task is already start!");
        }
        Set<TaskNode> allTaskNode = getAllTaskNode();
        taskNodeExecutor = Executors.newFixedThreadPool(allTaskNode.size());
        if (!CollectionUtils.isEmpty(allTaskNode)) {
            for (TaskNode taskNode : allTaskNode) {
                taskNodeExecutor.execute(() -> taskNode.execute());
            }
        }
        isEnable = true;
        // block queue poll
        taskPollExecutor.execute(() -> {
            while (true){
                if (!isEnable()) {
                    break;
                }
                tasks.forEach(task->{
                    BlockingQueue data = this.dataMap.get(task);
                    if(data == null || data.isEmpty()) return;
                    try {
                        firstNode.store(task, AbstractTaskNode.ROOT_NODE_NAME, data.take());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
        });
    }

    @Override
    public synchronized void stop() {
        closeEnable();
        Set<TaskNode> allTaskNode = getAllTaskNode();
        for (TaskNode taskNode : allTaskNode) {
            taskNode.stop();
        }
        taskPollExecutor.shutdown();
        taskNodeExecutor.shutdown();
    }

    @Override
    public Set<TaskNode> getAllTaskNode() {
        return taskNodes;
    }

    @Override
    public <T> void produce(String task, T data) throws InterruptedException {
        if(!tasks.contains(task)){
            tasks.add(task);
            firstNode.notifyTaskStatus(AbstractTaskNode.ROOT_NODE_NAME, task, TaskLife.ACTIVE);
        }
        BlockingQueue<T> taskData = null;
        if(dataMap.containsKey(task)){
            taskData = dataMap.get(task);
        }else {
            taskData = new ArrayBlockingQueue<>(1000);
            dataMap.put(task,taskData);
        }
        taskData.put(data);
    }

    @Override
    public boolean isFinish(String task) {
        Set<TaskNode> allTaskNode = getAllTaskNode();
        if (CollectionUtils.isEmpty(allTaskNode)) {
            return true;
        }
        for (TaskNode taskNode : allTaskNode) {
            if (!taskNode.taskIsFinished(task)) {
                return false;
            }
        }
        return true;
    }

    @Override
    public boolean isEnable() {
        return isEnable;
    }

    @Override
    public boolean notifyTaskFinish(String task) {
        logger.info(task +" data is over!!!");
        return firstNode.notifyTaskStatus(AbstractTaskNode.ROOT_NODE_NAME, task, TaskLife.FINISH);
    }


    private void closeEnable() {
        this.isEnable = false;
    }

}
